package com.mr;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
//import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
 * 把不规范、不能直接被hive建表的日志文件，在此进行处理，进行规范化
 */
public class ParseLogFile {

	static class LogMapper extends Mapper<LongWritable,Text,NullWritable,Text>{
		
		Text text=new Text();
		@Override
		public void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException {
			String orignValue=value.toString();
			StringBuilder strBuilder=new StringBuilder();
			
			String[] ipAndOther=orignValue.split("- -");
			strBuilder.append(ipAndOther[0].trim());
			strBuilder.append("    ");
			
			String[] timeAndOther=ipAndOther[1].split("\"");
			String time = timeAndOther[0].trim();
			time=time.substring(1, time.length()-2);
			strBuilder.append(time);
			strBuilder.append("    ");
			
			for(int i=1;i<timeAndOther.length;i++) {
				strBuilder.append(timeAndOther[i].trim());
				strBuilder.append("    ");
			}
			
			text.set(strBuilder.toString());
			context.write(NullWritable.get(), text);
		}//map
		
	}//map class
	
	/*static class LogReduce extends Reducer<LongWritable,Text,Text,Text>{
		
	}*/
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Path input=new Path(args[0]);
		Path output=new Path(args[1]);
		Configuration conf=new Configuration();
		FileSystem fs=FileSystem.get(conf);
		if(fs.exists(output)) {
			fs.delete(output, true);
		}
		
		Job job=Job.getInstance(conf, "parseLog");
		
		job.setJarByClass(ParseLogFile.class);
		job.setMapperClass(LogMapper.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(0);
		
		FileInputFormat.addInputPath(job, input);
		FileOutputFormat.setOutputPath(job, output);
		
		System.out.println(job.waitForCompletion(true)?0:1);
	}
}
